﻿#include <stdarg.h>
#include <unistd.h>
#include "threadTask.h"
#include "MemoryPoolObj.h"
#include "cmd.h"
#include "Net.h"
#include "define_protocol.h"
#include "PacketParse.h"

CThreadlTask::CThreadlTask()
{

	pthread_mutex_init(&m_pthreadMutex, nullptr);

	pthread_cond_init(&m_pthreadCond, nullptr);

	m_iRunningThreadNum = 0; 

	m_nTaskNum = 0;  

	m_iThreadNum = 0;
	m_shutdown = false;
}

CThreadlTask::~CThreadlTask()
{
	ClearTask();
}
void CThreadlTask::ClearTask()
{
	char* sTmpMempoint = nullptr;

	while (!m_MsgRecvQueue.empty())
	{
		sTmpMempoint = m_MsgRecvQueue.front();
		m_MsgRecvQueue.pop_front();
		MemPoolObj::GetMemoryPool()->Free((void*)sTmpMempoint);
		sTmpMempoint = nullptr;
	}
}
bool CThreadlTask::Start(int threadNum)
{

	m_iThreadNum = threadNum; 

	for (int i = 0; i < 4; ++i)
	{
		ThreadItem* pNew    = new ThreadItem(this);

		m_threadVector.push_back(pNew); 

		int err = pthread_create(&pNew->_Handle, nullptr, TaskFun, pNew);

		if (err != 0)
		{
			//创建线程有错
			printf("CThreadlTask::Create()创建线程%d失败，返回的错误码为%d! \n", i, err);
			return false;
		}
	} 
	std::vector<ThreadItem*>::iterator iter;
lblfor:
	for (iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
	{
		if ((*iter)->ifrunning == false)
		{
			usleep(100 * 1000);
			goto lblfor;
		}
	}
	return true;
}

void* CThreadlTask::TaskFun(void* threadData)
{
	ThreadItem* pThread = static_cast<ThreadItem*>(threadData);

	CThreadlTask* pThis = pThread->_pThis;

	int err;
	pthread_t tid = pthread_self(); 
	printf("TaskFun %d\n",(int)tid);

	while (1)
	{
		err = pthread_mutex_lock(&pThis->m_pthreadMutex);
		if (err != 0)
		{
			printf("CThreadlTask::TaskFun()pthread_mutex_lock()失败，返回的错误码为%d! \n", err);//有问题，要及时报告
		}

		while ((pThis->m_nTaskNum == 0) && pThis->m_shutdown == false)
		{
			if (pThread->ifrunning == false)
			{
				pThread->ifrunning = true;
			}
			pthread_cond_wait(&pThis->m_pthreadCond, &pThis->m_pthreadMutex); 
		}

		if (pThis->m_shutdown)
		{
			pthread_mutex_unlock(&pThis->m_pthreadMutex);	

			printf(" Break job---------------------------------------> %d\n", (int)pThis->m_nTaskNum);

			break;
		}
		
		char* jobbuf = pThis->m_MsgRecvQueue.front();   
		pThis->m_MsgRecvQueue.pop_front();                   
		--pThis->m_nTaskNum;

		LPSTRUC_MSG_HEADER pMsgHeader = (LPSTRUC_MSG_HEADER)jobbuf;
		
		sClient* p_Conn = pMsgHeader->p_Conn;
		--p_Conn->nRecvCount;
		err = pthread_mutex_unlock(&pThis->m_pthreadMutex);
		if (err != 0)
		{
			printf("CThreadlTask::TaskFun()pthread_cond_wait()失败，返回的错误码为%d! \n", err);//有问题，要及时报告
		} 
		printf("Job Size:%d all :%d \n", (int)p_Conn->nRecvCount, pThis->m_nTaskNum);
	  
		CPacketParse::GetInstance()->ParseCmd(jobbuf);


		 

		g_pServer->CloseConnection(p_Conn);             

		MemPoolObj::GetMemoryPool()->Free((void*)jobbuf);
		jobbuf = nullptr;
	}
	printf("Job TaskFun    Exit... \n");
	return (void*)0;
}
void CThreadlTask::Stop()
{
	while (m_nTaskNum > 0)
	{
		printf("wait: m_iRecvMsgQueueCount =%d\n", (int)m_nTaskNum);
		pthread_cond_broadcast(&m_pthreadCond);
		usleep(1000);
	}

	if (m_shutdown == true)
	{
		return;
	}

	m_shutdown = true;
	

	int err = pthread_cond_broadcast(&m_pthreadCond);
	if (err != 0)
	{
		printf("CThreadlTask::StopAll()中pthread_cond_broadcast()失败，返回的错误码为%d! \n", err);
		return;
	}
	printf("Job CThreadlTask StopAll pthread_cond_broadcast\n");
	
	std::vector<ThreadItem*>::iterator iter;
	for (iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
	{
		pthread_join((*iter)->_Handle, nullptr); 
	}

	pthread_mutex_destroy(&m_pthreadMutex);
	
	pthread_cond_destroy(&m_pthreadCond);

	for (iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
	{
		if (*iter)
		{
			delete* iter;
		}
	}

	m_threadVector.clear();

	return;
}
void CThreadlTask::AddTask(char* buf)
{
	int err = pthread_mutex_lock(&m_pthreadMutex);
	if (err != 0)
	{
		printf("CThreadlTask::AddTask()pthread_mutex_lock()失败，返回的错误码为%d! \n", err);
	}
	m_MsgRecvQueue.push_back(buf);
	
	LPSTRUC_MSG_HEADER pMsgHeader = (LPSTRUC_MSG_HEADER)buf;                  //消息头

    sClient *p_Conn = pMsgHeader->p_Conn; 

	if (!p_Conn)
	{
		err = pthread_mutex_unlock(&m_pthreadMutex);
		return;
	}

	++m_nTaskNum;  
	++p_Conn->nRecvCount;
	err = pthread_mutex_unlock(&m_pthreadMutex);

	

	if (err != 0)
	{
		printf("CThreadlTask::AddTask()pthread_mutex_unlock()失败，返回的错误码为%d! \n", err);
	}
	Wakeup();
	return;
}
void CThreadlTask::Wakeup()
{
	int err = pthread_cond_signal(&m_pthreadCond); 
	if (err != 0)
	{
		printf("CThreadlTask::Call()中pthread_cond_signal()失败，返回的错误码为%d! \n", err);
	}
	return;
}
